小王|私信_Kettle实现ES到ES循环增量抽取

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Kettle实现ES到ES循环增量抽取相关的知识，希望对你有一定的参考价值。

主页:​​小王叔叔的博客​​欢迎来访

支持:点赞​收藏​关注

本博客内容，实践前，请先逐一浏览，然后再逐一学习

1、效果

​

2、实现

2.1 创建数据库

见 ​​Kettle安装使用​​

2.2 创建作业

2.2.1 初始化变量：设置变量，通过变量实现作业的循环更新初始值

​

parent_job.setVariable("isContinue", "1");
parent_job.setVariable("lastUpdateTime", "");
true;

2.2.2 创建核心转换

【见2.3】

2.2.3 写日志记录

​

isCOntinue= $isContinue-------------------
lastUpdateTime = $lastUpdateTime===============

2.2.4 设置循环

​

通过【2.2】中设置，可以将基本循环抽取动作的作业可以实现循环。

2.3 创建转换：关键处！！！

思路：

1.通过mysql中kettle业务抽取的时间备用表，进行设置最后一次修改更新时间。

2.设置基本循环单次抽取的条数，和基本抽取的json格式

3.设置抽取的数据源

4.解析抽取后的es中内置的hits-source的相关结构

5.成功解析之后，将抽取到的数据进行入库，同时变量获取最新的更新时间保存到MySQL中，便于下次更新使用

2.3.1 选择数据源

select round(unix_timestamp(timetable_dev.modify_time)*1000) as modifyTime, 1 as isContinue
from es_kettle.timetable_dev WHERE index_name = sta_resource_operation

2.3.2 更新常量

"from":0,"size":10,"query":"bool":"filter":["bool":"must":["range":"last_update_time":"from":startTime,"to":null,"include_lower":true,"include_upper":true,"boost":1],"adjust_pure_negative":true,"boost":1],"adjust_pure_negative":true,"boost":1,"sort":["last_update_time":"order":"asc"]

2.3.4 参数替换

​2.3.5 设置数据源基本请求信息

​2.3.6 配置解析hits结构

​2.3.7 配置解析的结构

​2.3.8 筛选结构

​2.3.9 最后如ES库

​2.3.10 根据时间设置循环

​

执行SQL脚本：

update es_kettle.timetable_dev set modify_time = FROM_UNIXTIME(?, %Y-%m-%d %H:%i:%S) where index_name = sta_resource_operation

以上就是ES通过作业，转换进行抽取到新的ES结果

3、注意事项

3.1）设置对应字段

3.2）组件之间的关联性

4、最后完成效果

​

转载声明：本文为博主原创文章，未经博主允许不得转载

⚠️注意 ~